package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}

object UserActionHotProductAnalysis {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf和SparkContext
    val conf = new SparkConf()
      .setAppName("Hot Product Analysis")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 1. 读取用户行为日志文件
    val actionRDD = sc.textFile("/opt/code/input/user_visit_action.txt")

    // 2. 解析数据并提取产品信息
    val productActionRDD = actionRDD.flatMap(line => {
      val fields = line.split("_")
      
      // 确保数据格式正确
      if (fields.length >= 13) {
        // 提取点击、下单、支付的产品ID
        val clickProducts = fields(10).split(",")
        val orderProducts = fields(11).split(",")
        val payProducts = fields(12).split(",")

        // 生成(产品ID, (点击数, 下单数, 支付数))的元组
        (clickProducts.map((_, (1, 0, 0))) ++
         orderProducts.map((_, (0, 1, 0))) ++
         payProducts.map((_, (0, 0, 1))))
      } else {
        Nil
      }
    })

    // 3. 聚合产品的点击、下单、支付数据
    val productStatsRDD = productActionRDD
      .reduceByKey((a, b) => (
        a._1 + b._1,  // 点击数累加
        a._2 + b._2,  // 下单数累加
        a._3 + b._3   // 支付数累加
      ))

    // 4. 排序并获取热度前10的产品
    val topProducts = productStatsRDD
      .map { case (productId, (clickCount, orderCount, payCount)) => 
        (productId, clickCount, orderCount, payCount)
      }
      .sortBy(x => (-x._2, -x._3, -x._4))  // 按点击数降序，再按下单数，再按支付数
      .take(10)

    // 5. 打印结果
    println("热度Top 10产品:")
    topProducts.foreach { case (productId, clickCount, orderCount, payCount) =>
      println(f"产品ID: $productId, 点击数: $clickCount, 下单数: $orderCount, 支付数: $payCount")
    }

    // 6. 保存结果到文件（可选）
    val outputPath = "/opt/code/output/hot_products.txt"
    sc.parallelize(topProducts.toSeq)
      .map { case (productId, clickCount, orderCount, payCount) => 
        s"$productId,$clickCount,$orderCount,$payCount"
      }
      .saveAsTextFile(outputPath)

    println(s"\n结果已保存到: $outputPath")

    // 暂停以便查看Spark UI
    Thread.sleep(300000)

    sc.stop()
  }
} 